#pragma once
#include <iostream>
#include <string>
#include <thread>
#include <vector>
#include <unordered_map>
#include <algorithm>
#include <cstring>
#include <unistd.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <functional>
#include <sys/epoll.h>
#include <mutex>
#include <sys/eventfd.h>
#include <memory>
#include <sys/timerfd.h>
#include <condition_variable>
#include <any>


#define DBG 0
#define INF 1
#define ERR 2
#define DBG_LEVEL DBG

#define LOG(level, format, ...)\
    do\
    {\
        if (level >= DBG_LEVEL)\
        {\
            time_t curtime = time(nullptr);\
            struct tm totime;\
            localtime_r(&curtime, &totime);\
            char readytime[32];\
            strftime(readytime, 31, "%Y:%m:%d-%H:%M%S", &totime);\
            std::cout << "[" << std::this_thread::get_id() << "]";\
            fprintf(stdout, "[%s][%s][%d] " format "\n", readytime, __FILE__, __LINE__, ##__VA_ARGS__);\
        }\
    }while(0)

#define DBG_LOG(format, ...) LOG(DBG, format, ##__VA_ARGS__)
#define INF_LOG(format, ...) LOG(INF, format, ##__VA_ARGS__)
#define ERR_LOG(format, ...) LOG(ERR, format, ##__VA_ARGS__)

#define BUFFERDEFAULTSIZE 65535


class Buffer
{
private:
    std::vector<char> _buffer;
    int _reader_index;
    int _writer_index;


public:
    Buffer()
        :_buffer(BUFFERDEFAULTSIZE),
        _writer_index(0),
        _reader_index(0)
    {}

    void *Begin()
    {
        return &_buffer[0];
    }

    void *ReadPosition()
    {
        return &_buffer[_reader_index];
    }

    void *WritePosition()
    {
        return &_buffer[_writer_index];
    }

    int ReadAbleSize()
    {
        return _writer_index - _reader_index;
    }
    
    int WriteAbleSize()
    {
        return _buffer.size() - _writer_index;
    }

    bool MoveReader(int len)
    {
        if(len > ReadAbleSize()) 
        {
            ERR_LOG("MoveReader error");
            return false;
        }
        _reader_index += len;
        return true;
    }

    bool MoveWriter(int len)
    {
        if (len > WriteAbleSize())
        {
            ERR_LOG("MoveWriter error");
            return false;
        }
        _writer_index += len;
        return true;
    }

    void EnsureAbleWrite(int len)
    {
        if (len <= WriteAbleSize()) return;
        else if (len <= WriteAbleSize() + _reader_index)
        {
            //将中间部分可读数据移动到开头位置， 然后将read指针移动到0位置， write指针移动到0 + readablesize(移动前)
            int readablesize = ReadAbleSize();
            std::copy((char*)ReadPosition(), (char*)ReadPosition() + ReadAbleSize(), (char*)Begin());
            _reader_index = 0;
            _writer_index = readablesize;
        }
        else
        {
            //空间不足， 直接二倍 + len扩容就行了, 指针不需要动
            int size = len + 2 * _buffer.size();
            _buffer.resize(size);
        }
    }

    int Read(void *data, int len)
    {
        int minlen = std::min(len, ReadAbleSize());
        std::copy((char*)ReadPosition(), (char*)ReadPosition() + minlen, (char*)data);
        return minlen; 
    }

    int Write(void *data, int len)
    {
        EnsureAbleWrite(len);
        std::copy((char*)data, (char*)data + len, (char*)WritePosition());
        return len;
    }

    int ReadAndPush(void *data, int len)
    {
        int ret = Read(data, len);
        if (!MoveReader(ret))
        {
            return -1;
        }
        return ret;
    }

    int WriteAndPush(void *data, int len)
    {
        int ret = Write(data, len);
        if (!MoveWriter(ret))
        {
            return -1;
        }
        return ret;
    }

    std::string ReadAsString(int len)
    {
        std::string str;
        str.resize(len);

        int ret = Read(&str[0], str.size());
        return str;
    }

    std::string ReadAsStringAndPush(int len)
    {
        std::string ret = ReadAsString(len);
        int size = ret.size();
        MoveReader(size);
        return ret;
    }

    char *FindCRLF()
    {
        char *res = (char*)memchr(ReadPosition(), '\n', ReadAbleSize());
        return res;
    }

    std::string GetLine()
    {
        char *end = FindCRLF();
        if (end == nullptr)
        {
            return "";
        }
        int size = (end - (char*)ReadPosition() + 1);
        return ReadAsString(size);
    }

    std::string GetLineAndPush()
    {
        std::string ret = GetLine();
        int size = ret.size();
        MoveReader(size);
        return ret;
    }

    int WriteAsString(std::string str)
    {
        int ret = Write((void*)str.c_str(), str.size());
        return ret;
    }

    int WriteAsStringAndPush(std::string str)
    {
        int ret = WriteAsString(str);
        if (!MoveWriter(ret))
        {
            return -1;
        }
        return ret;
    }

    int WriteBuffer(Buffer &buf)
    {
        int ret = Write(buf.ReadPosition(), buf.ReadAbleSize());
        return ret;
    }

    int WriteBufferAndPush(Buffer &buf)
    {
        int ret = WriteBuffer(buf);
        if (!MoveWriter(ret))
        {
            return -1;
        }
        return ret;
    }

};

class Socket
{
private:
    int _sockfd;

public:
    Socket(){}
    Socket(int sockfd)
        :_sockfd(sockfd)
    {}

    int Fd()
    {
        return _sockfd;
    }

    bool Create()
    {   
        _sockfd = socket(AF_INET, SOCK_STREAM, 0); 
        if (_sockfd < 0)
        {
            ERR_LOG("create sockfd failed");
            return false;
        }
        return true;
    }

    bool Bind(uint16_t port, std::string ip)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;    //IPv4
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(addr);
        int ret = bind(_sockfd, (const struct sockaddr*)&addr, len);
        if (ret < 0)
        {
            ERR_LOG("bind error");
            return false;
        }
        return true;
    }

    bool Listen()
    {
        int ret = listen(_sockfd, 10);
        if (ret < 0)
        {
            ERR_LOG("listen error");
            return false;
        }   
        return true;
    }

    bool Connect(uint16_t port, std::string ip = "127.0.0.1")
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;    //IPv4
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(addr);
        int ret = connect(_sockfd, (struct sockaddr*)&addr, len);
        if (ret < 0)
        {
            ERR_LOG("connect error");
            return false;
        }
        return true;
    }

    int Accept()
    {
        int ret = accept(_sockfd, nullptr, nullptr);
        if (ret < 0)
        {
            return -1;
        }
        return ret;
    }

    void SetNonBlock()
    {
        int fl = fcntl(_sockfd, F_GETFL, 0);
        int ret = fcntl(_sockfd, F_SETFL, fl | SOCK_NONBLOCK);
        if (ret < 0)
        {
            int cnt = 3;
            while (cnt--)
            {
                ERR_LOG("set NONBLOCK error");
                sleep(1);
            }
            return;
        }
    }

    int Recv(void *data, int len, int flag = 0)
    {
        int ret = recv(_sockfd, data, len, flag);
        if (ret < 0)
        {
            if (errno == EINTR || errno == EAGAIN)
            {
                return 0;
            }
            ERR_LOG("client breakdown, sockfd : %d", _sockfd);
            return -1;
        }
        return ret;
    }

    int Send(const void *data, int len, int flag = 0)
    {
        int ret = send(_sockfd, data, len, flag);
        if (ret < 0)
        {
            if (errno == EINTR || errno == EAGAIN)
            {
                return 0;
            }
            ERR_LOG("process unwork, send error");
            sleep(1);
            return -1;
        }
        return ret;
    }

    int NonBlockRecv(void *data, int len)
    {
        int ret = Recv(data, len, MSG_DONTWAIT);
        return ret;
    }

    int NonBlockSend(void *data, int len)
    {
        int ret = Send(data, len, MSG_DONTWAIT);
        return ret;
    }
    
    bool CreateServer(uint16_t port, std::string ip = "0.0.0.0", int blockflag = 0)
    {
        if (!Create())
        {
            return false;
        }
        if (blockflag == 1) SetNonBlock();
        if (!Bind(port, ip))
        {
            return false;
        }
        if (!Listen())
        {
            return false;
        }
        Reuse();
        return true;
    }    

    bool CreateConnection(uint16_t port, std::string ip = "0.0.0.0")
    {
        if (!Create()) return false;
        if (!Connect(port, ip)) return false;

        return true;
    }   

    void Reuse()
    {
        int val = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (void*)&val, sizeof(int));
        val = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, (void*)&val, sizeof(int));
    }

    void Close()
    {
        if (_sockfd != -1)
        {
            close(_sockfd);
            DBG_LOG("close sockfd : %d", _sockfd);
            _sockfd = -1;
        }
    }

};


class EventLoop;
class Poller;

class Channel
{
public:
    using EventCallBack = std::function<void()>;
private:
    int _sockfd;
    EventLoop *_loop;    
    int _event;         //该文件描述符要监控的事件
    int _revent;        //该文件描述符已经就绪的事件

    EventCallBack HandleRead;
    EventCallBack HandleWrite;
    EventCallBack HandleError;
    EventCallBack HandleClose;
    EventCallBack HandleAny;

public:
    Channel(EventLoop *loop, int sockfd)
        :_loop(loop), _sockfd(sockfd),
        _event(0), _revent(0)
    {}


    void SetHandleRead(const EventCallBack &cb)
    {
        HandleRead = cb;
    }
    void SetHandleWrite(const EventCallBack &cb)
    {
        HandleWrite = cb;
    }
    void SetHandleError(const EventCallBack &cb)
    {
        HandleError = cb;
    }
    void SetHandleClose(const EventCallBack &cb)
    {
        HandleClose = cb;
    }
    void SetHandleAny(const EventCallBack &cb)
    {
        HandleAny = cb;
    }

    void SetRevent(int event)    //设置实际就绪的事件
    {
        _revent = event;
    }

    int Fd()
    {
        return _sockfd;
    }

    int GetEvent()
    {
        return _event;
    }

    bool AbleRead()
    {
        return _event & EPOLLIN;
    }
    bool AbleWrite()
    {
        return _event & EPOLLOUT;
    }

    void EnableRead()
    {
        _event |= EPOLLIN;
        UpData();

    }
    void EnableWrite()
    {
        _event |= EPOLLOUT;
        UpData();

    }

    void DisableRead()
    {
        _event &= ~EPOLLIN;
        UpData();

    }
    void DisableWrite()
    {
        _event &= ~EPOLLOUT;
        UpData();
    }

    void DisAbleAll()
    {
        _event = 0;
        UpData();
    }

    void UpData();
    void Remove();

    void HandleEvent()
    {

        if (_revent & EPOLLIN || _revent & EPOLLHUP || _revent & EPOLLPRI)
        {
            if (HandleRead)
            {
                HandleRead();
            }
        }
        if (_revent & EPOLLOUT)
        {
            if (HandleWrite)
            {
                HandleWrite();
            }
        }
        if (_revent & EPOLLRDHUP || _revent & EPOLLHUP)
        {
            if (HandleClose)
            {
                HandleClose();
            }
        }
        if (_revent & EPOLLERR)
        {
            if (HandleError)
            {
                HandleError();
            }
        }

        if (HandleAny)
        {
            HandleAny();
        }
    }
};  

#define MAXEPOLLEVENTSIZE 1024

class Poller
{
private:
    int _epfd;
    struct epoll_event _revents[MAXEPOLLEVENTSIZE];
    std::unordered_map<int, Channel *> _epolls_map; 

private:
    bool Updata(Channel *chn, int op)
    {
        int fd = chn->Fd();
        struct epoll_event ev;
        ev.data.fd = fd;
        ev.events = chn->GetEvent();

        int ret = epoll_ctl(_epfd, op, fd, &ev);
        if (ret < 0)
        {
            ERR_LOG("set epoll error, sockfd: %d; op : %d", ev.data.fd, op);
            return false;
        }
        return true;
    }

    bool HasChannel(Channel *chn)
    {
        int fd = chn->Fd();
        auto it = _epolls_map.find(fd);
        if (it == _epolls_map.end())
        {
            return false;
        }
        return true;
    }

public:
    Poller()
    {
        _epfd = epoll_create(10000);
        if (_epfd < 0)
        {
            ERR_LOG("create epoll failed");
            abort();
        }
    }

    bool UpdateEvent(Channel *chn)
    {
        if (!HasChannel(chn))
        {
            if (!Updata(chn, EPOLL_CTL_ADD))
            {
                return false;
            }
            int fd = chn->Fd();
            _epolls_map[fd] = chn;
        }
        else
        {

            if (!Updata(chn, EPOLL_CTL_MOD))
            {
                return false;
            }
        }
        return true;
    }

    bool RemoveEvent(Channel *chn)
    {
        int fd = chn->Fd();
        auto it = _epolls_map.find(fd); //从epoll模型里面找fd
        if(it != _epolls_map.end())
        {
            _epolls_map.erase(it);
        }


        if (!Updata(chn, EPOLL_CTL_DEL))
        {
            return false;
        }

        return true;
    }

    int Poll(std::vector<Channel *> *actives)
    {
        int ret = epoll_wait(_epfd, _revents, MAXEPOLLEVENTSIZE, -1);
        if (ret < 0)
        {
            if (errno == EINTR || errno == EAGAIN)
            {
                return 0;
            }

            ERR_LOG("epoll_wait error : %s", strerror(errno));
            return -1;
        }
        else if (ret == 0)
        {
            return ret;
        }
        else
        {
            for (int i = 0; i < ret; i++)
            {
                struct epoll_event ev = _revents[i];
                int fd = ev.data.fd;
                int revent = ev.events;

                auto it = _epolls_map.find(fd);
                if (it == _epolls_map.end())
                {
                    ERR_LOG("Remove and Poll not serialization : %d", fd);
                    abort();  
                }

                _epolls_map[fd]->SetRevent(revent);
                actives->push_back(_epolls_map[fd]);
            }
        }

        return ret;
        
    }
};


using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;
class TimeTask
{
public:
    int _task_id;
    int _timeout;
    TaskFunc _task;
    ReleaseFunc _re_task;
    bool _cancel;

public:
    TimeTask(int id, int delay, const TaskFunc& ftor)
        :_task_id(id),
        _timeout(delay),
        _task(ftor),
        _cancel(false)
    {}

    void SetReleaseCallBack(const ReleaseFunc &cb)
    {
        _re_task = cb;
    }

    void cancel()
    {
        _cancel = true;
    }

    ~TimeTask()
    {
        if (_cancel == false)
        {
            _task();
        }
        _re_task();
    }

};

class TimeWheel
{
public:
    using SPtr = std::shared_ptr<TimeTask>;
    using WPtr = std::weak_ptr<TimeTask>;
private:
    EventLoop *_loop;

    int _tick;
    int _capacity;
    std::vector<std::vector<SPtr>> _time_wheel;
    std::unordered_map<int, WPtr> _map_task;

    int _timerfd;
    Channel _timer_channel;

private:

    void Releasetask(int id)
    {
        auto it = _map_task.find(id);
        if (it != _map_task.end())
        {
            _map_task.erase(it);
        }
    }

    int CreateTimerfd()
    {
        int timerfd = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
        if (timerfd < 0)
        {
            ERR_LOG("create timer fd error, exit thread");
            pthread_exit(0);
        }
        DBG_LOG("create timerfd : %d", timerfd);
        struct itimerspec setval;
        setval.it_interval.tv_sec = 1;
        setval.it_interval.tv_nsec = 0;
        setval.it_value.tv_sec = 1;
        setval.it_value.tv_nsec = 0;

        int ret = timerfd_settime(timerfd, 0, &setval, nullptr);
        if (ret < 0)
        {
            ERR_LOG("set timerfd time error, thread exit");
            pthread_exit(0);
        }
        return timerfd;
    }

    void OneStep()
    {
        _time_wheel[_tick].clear();
        _tick = ((_tick + 1) % _capacity);        
    }

    int TimerHandleRead()
    {
        uint64_t val = 0;
        int ret = read(_timerfd, &val, sizeof(val));
        if (ret < 0)
        {
            ERR_LOG("timerfd read error");
            _timer_channel.DisableRead();
            close(_timerfd);
            // loop->Destroy();   //这个是线程销毁函数， 当一个线程崩溃的时候， 这个线程内部对所有连接的监控都应该释放。
            return -1;
        }
        return val;
    }

    void OnTime()
    {
        int ret = TimerHandleRead();
        for (int i = 0; i < ret; i++)
        {
            OneStep();
        }
    }



    void AddTimerInLoop(int task_id, int delay, const TaskFunc &task)
    {
        SPtr ptr = std::make_shared<TimeTask>(task_id, delay, task);
        ptr->SetReleaseCallBack(std::bind(&TimeWheel::Releasetask, this, task_id));

        WPtr fakeptr(ptr);

        if (_map_task.count(task_id))  //说明已经存在了
        {
            return;
        }

        int pos = ((_tick + delay) % _capacity);
        _time_wheel[pos].push_back(ptr);
        _map_task[task_id] = fakeptr;
    }    

    void FlushTimerInLoop(int task_id)
    {
        auto it = _map_task.find(task_id);
        if (it == _map_task.end())
        {
            ERR_LOG("timer has error");
            return;
        }
        SPtr ptr = (it->second).lock();
        int delay = ptr->_timeout;
        int pos = (_tick + delay) % _capacity;
        _time_wheel[pos].push_back(ptr);
    }

    void CancelTaskInLoop(int id)
    {
        auto it = _map_task.find(id);
        if (it == _map_task.end())
        {
            return;
        }
        SPtr ptr = (it->second).lock();
        if (ptr != nullptr) ptr->cancel();   
    }

public:
    TimeWheel(EventLoop *loop)
        :_loop(loop),
        _tick(0),
        _capacity(60),
        _time_wheel(_capacity),
        _timerfd(CreateTimerfd()),
        _timer_channel(loop, _timerfd)
    {
        _timer_channel.SetHandleRead(std::bind(&TimeWheel::OnTime, this));
        _timer_channel.EnableRead();
    }

    bool HasTimer(int id)
    {
        auto it = _map_task.find(id);
        if (it != _map_task.end())
        {
            return true;
        }
        return false;
    }

    void CancelTask(int id);

    void AddTimer(int task_id, int delay, const TaskFunc &task);

    void FlushTimer(int task_id);

};


class EventLoop
{
public:
    using FuncTask = std::function<void()>;
private:
    std::thread::id _tid;
    std::mutex _mutex;
    Poller _poller;
    int _eventfd;
    Channel _event_channel;
    std::vector<FuncTask> _tasks;
    TimeWheel _timewheel;

private:
    int CreateEventfd()
    {
        int efd = eventfd(0, FD_CLOEXEC | O_NONBLOCK);
        if (efd < 0)
        {
            ERR_LOG("create eventfd error, thread id <------");
            return -1;
        }
        DBG_LOG("create eventfd : %d", efd);
        return efd;
    }

    void WeakUpEventfd()
    {
        uint64_t val = 0;
        int ret = write(_eventfd, &val, sizeof(val));
        if (ret < 0)
        {
            ERR_LOG("eventfd write error, thread exit");
            _event_channel.Remove();
            close(_eventfd);
            pthread_exit(nullptr);
        }
    }

    void ReadEventfd()
    {
        uint64_t val = 0;
        int ret = read(_eventfd, &val, sizeof(val));
        if (ret < 0)
        {
            ERR_LOG("eventfd read error, thread exit");
            _event_channel.Remove();
            close(_eventfd);
            pthread_exit(nullptr);
        }
    }

    void RunAllTask()
    {
        std::vector<FuncTask> tmptasks;
        {
            std::unique_lock<std::mutex> _lock(_mutex);   
            _tasks.swap(tmptasks);                          //对于_tasks的交换操作需要加锁， 因为_tasks同时被主线程加入task
        }                                                   //和子线程拿走task时会发生冲突。 所以_task是一种临界资源
        for (auto &e : tmptasks)     //对于tmp_tasks来说， 就是独属于子线程自己的资源了， 因为Poll保证由threadentry调用
        {
            e();     
        }
    }

    bool IsInLoop()
    {
        return _tid == std::this_thread::get_id();
    }


    void UpdateEventInLoop(Channel *chn)
    {
        _poller.UpdateEvent(chn);
    }
    void RemoveEventInLoop(Channel *chn)
    {
        _poller.RemoveEvent(chn);
    }
public:
    EventLoop()
        :_tid(std::this_thread::get_id()),
        _eventfd(CreateEventfd()),
        _event_channel(this, _eventfd), 
        _timewheel(this)
    {
        _event_channel.SetHandleRead(std::bind(&EventLoop::ReadEventfd, this));
        _event_channel.EnableRead();
    }

    void QueueInLoop(const FuncTask &functor)
    {
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.push_back(functor);
        }
        WeakUpEventfd();             //走到这， 说明是主线程，此时子线程可能需要唤醒
    }

    void RunInLoop(const FuncTask &functor)
    {
        if (IsInLoop())
        {
            functor();
            return;      //走到这，说明是子线程，此时不需要唤醒
        }
        
        QueueInLoop(functor);
    }

    void UpdateEvent(Channel *chn)
    {
        RunInLoop(std::bind(&EventLoop::UpdateEventInLoop, this, chn));   //bind函数时， 类内成员函数的this指针不能忘记。
    }

    void RemoveEvent(Channel *chn)
    {
        RunInLoop(std::bind(&EventLoop::RemoveEventInLoop, this, chn));
    }

    void CancelTask(int id)
    {
        _timewheel.CancelTask(id);
    }

    bool HasTimer(int id)
    {
        return _timewheel.HasTimer(id);
    }

    void AddTimer(int id, int delay, const TaskFunc &cb)
    {
        _timewheel.AddTimer(id, delay, cb);
    }

    void FlushTimer(int id)
    {
        _timewheel.FlushTimer(id);
    }

    void Loop()
    {
        while (1)
        {
            std::vector<Channel *> actives;
            _poller.Poll(&actives);
            for (auto &e : actives)
            {
                e->HandleEvent();
            }

            RunAllTask();      //监控完之后， 就把任务池里面的所有任务都拿出来执行。 
        }
    }

};


void Channel::UpData()
{
    _loop->UpdateEvent(this);
}

void Channel::Remove()
{
    _loop->RemoveEvent(this);
}


void TimeWheel::CancelTask(int id)
{
    _loop->RunInLoop(std::bind(&TimeWheel::CancelTaskInLoop, this, id));
}    

void TimeWheel::AddTimer(int task_id, int delay, const TaskFunc &task)
{
    _loop->RunInLoop(std::bind(&TimeWheel::AddTimerInLoop, this, task_id, delay, task));
}

void TimeWheel::FlushTimer(int task_id)
{
    _loop->RunInLoop(std::bind(&TimeWheel::FlushTimerInLoop, this, task_id));
}


enum class ConnectState
{
    DISCONNECTED,
    CONNECTING,
    CONNECTED,
    DISCONNECTING
};

class Connection;
using PtrConn = std::shared_ptr<Connection>;

class Connection : public std::enable_shared_from_this<Connection>
{
public:
    using MessageCallBack = std::function<void(const PtrConn &, Buffer *)>;
    using ConnectCallBack = std::function<void(const PtrConn &)>;
    using CloseCallBack = std::function<void(const PtrConn &)>;
    using AnyCallBack = std::function<void(const PtrConn &)>;

private:
    int _conn_id;
    EventLoop *_loop;
    int _sockfd;
    Socket _socket;
    Channel _channel;

    Buffer _in_buffer;
    Buffer _out_buffer;
    ConnectState _state;
    bool _inactiveconnectiondestroy;

    std::mutex _mutex;

    std::any _content;
    MessageCallBack _msg_call;
    ConnectCallBack _conn_call;
    AnyCallBack _any_call;
    CloseCallBack _close_call;
    CloseCallBack _serverclose_call;

private:
    void HandleRead()
    {
        char buffer[65535] = { 0 };
        int ret = _socket.Recv(buffer, sizeof(buffer));
        if (ret < 0)
        {
            ReleaseConnection();
            return;
        }
        else if (ret == 0)
        {
            DBG_LOG("recv 0 characters, close sockfd : %d", _sockfd);
            ShutDown();
        }
        else
        {
            _in_buffer.WriteAndPush(buffer, ret);
            if (_msg_call) _msg_call(shared_from_this(), &_in_buffer);
        }
    }

    void HandleWrite()
    {
        int ret = _socket.NonBlockSend(_out_buffer.ReadPosition(), _out_buffer.ReadAbleSize());
        if (ret < 0)
        {
            ERR_LOG("Buffer break down, Connection relese");
            ReleaseConnection();
            return;
        }
        if (ret == 0)
        {
            _channel.DisableWrite();
        }
        _out_buffer.MoveReader(ret);
        if (_out_buffer.ReadAbleSize() == 0) _channel.DisableWrite();
        if (_state == ConnectState::DISCONNECTING)     //一般走不到这里， 如果处于半关闭， 说明在recv状态时关闭客户端， _state置为了半关闭
        {                                //， 然后来发送最后一次确认应答
            ReleaseConnection();
        }
    }

    void HandleError()
    {
        HandleClose();
    }

    void HandleClose()
    {
        ReleaseConnection();
    }

    void HandleAny()
    {
        if (_inactiveconnectiondestroy) _loop->FlushTimer(_conn_id);
    }

    void EstablishedInLoop()
    {
        if (_state != ConnectState::CONNECTING)
        {
            ERR_LOG("create connection logical is error, exit");
            exit(0);
        }
        _state = ConnectState::CONNECTED;
        _channel.EnableRead();
        if (_conn_call) _conn_call(shared_from_this());
    }

    void SendInLoop(void *data, int len)
    {   
        int ret = _out_buffer.WriteAndPush(data, len);
        if (ret < 0)
        {
            ERR_LOG("outbuffer resize error, connection close");
            ReleaseConnection();
            return;
        }
        _channel.EnableWrite();
    }

    void EnableInActiveDestroyInLoop(int id, int delay, const TaskFunc &cb)
    {
        _inactiveconnectiondestroy = true;

        if (!_loop->HasTimer(id))
        {
            _loop->AddTimer(id, delay, cb);
        }
        else
        {
            _loop->FlushTimer(id);
        }
    }

    void CancelInActiveDestroyInLoop(int id)
    {
        _inactiveconnectiondestroy = false;
        _loop->CancelTask(id);
    }

    void ReleaseConnectionInLoop()
    {
        _state = ConnectState::DISCONNECTED;
        _channel.Remove();
        _socket.Close();
        if (_inactiveconnectiondestroy)
        {
            CancelInActiveDestroy(_conn_id);
        }

        if (_close_call) _close_call(shared_from_this());
        if (_serverclose_call) _serverclose_call(shared_from_this());
    }

    void ShutDownInLoop()
    {
        _state = ConnectState::DISCONNECTING;

        if (_out_buffer.ReadAbleSize() == 0)
        {
            ReleaseConnection();
            return;
        }
        _channel.EnableWrite();
    }

public:

    Connection(EventLoop *loop, int conn_id, int sockfd)
        :_loop(loop), _conn_id(conn_id),
        _sockfd(sockfd), _socket(_sockfd), 
        _channel(_loop, _sockfd),
        _state(ConnectState::CONNECTING),
        _inactiveconnectiondestroy(false)
    {
        _channel.SetHandleRead(std::bind(&Connection::HandleRead, this));
        _channel.SetHandleWrite(std::bind(&Connection::HandleWrite, this));
        _channel.SetHandleError(std::bind(&Connection::HandleError, this));
        _channel.SetHandleClose(std::bind(&Connection::HandleClose, this));
        _channel.SetHandleAny(std::bind(&Connection::HandleAny, this));
    }

    void SetMessageCallBack(const MessageCallBack &cb)
    {
        _msg_call = cb;
    }

    
    void SetConnectCallBack(const ConnectCallBack &cb)
    {
        _conn_call = cb;
    }

    void SetAnyCallBack(const AnyCallBack &cb)
    {
        _any_call = cb;
    }

    void SetCloseCallBack(const CloseCallBack &cb)
    {
        _close_call = cb;
    }

    void SetServerCloseCallBack(const CloseCallBack &cb)
    {
        _serverclose_call = cb;
    }

    void EnableInActiveDestroy(int id, int delay, const TaskFunc &cb)
    {
        _loop->RunInLoop(std::bind(&Connection::EnableInActiveDestroyInLoop, this, id, delay, cb));
    }

    int GetFd()
    {
        return _sockfd;
    }

    int GetId()
    {
        return _conn_id;
    }

    std::any &GetContent()
    {
        return _content;
    }

    void SetContent(std::any content)
    {

        _content = content;
    }

    void Established()
    {
        _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this));
    }

    void Send(void *data, int len)
    {
        _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, data, len));
    }

    void CancelInActiveDestroy(int id)
    {
        _loop->RunInLoop(std::bind(&Connection::CancelInActiveDestroyInLoop, this, id));
    }

    void ShutDown()
    {
        _loop->RunInLoop(std::bind(&Connection::ShutDownInLoop, this));
    }

    void ReleaseConnection()
    {
        _loop->QueueInLoop(std::bind(&Connection::ReleaseConnectionInLoop, this));
    }

};

class Acceptor
{
public:
    using ConnectCallBack = std::function<void(int)>;
private:
    EventLoop *_baseloop;
    Socket _socket;
    Channel _accept_channel;
    
    ConnectCallBack _conn_call;

private:    
    void HandleRead()
    {
        int newfd = _socket.Accept();
        if (newfd < 0)
        {
            ERR_LOG("recv connection error, server or client undefine");
            return;
        }
        if (!_conn_call)
        {
            ERR_LOG("acceptor connectioncallback unset, exit");
            exit(-1);
        }
        _conn_call(newfd);
    }

public:
    Acceptor(EventLoop *loop, uint16_t port)
        :_baseloop(loop),
        _socket(CreateSocket(port)),
        _accept_channel(_baseloop, _socket.Fd())
    {
        _accept_channel.SetHandleRead(std::bind(&Acceptor::HandleRead, this));
    }

    int Fd()
    {
        return _socket.Fd();
    }

    int CreateSocket(uint16_t port)
    {
        if (!_socket.CreateServer(port))
        {
            ERR_LOG("create acceptor error, exit");
            exit(-1);
        }
        int ret = _socket.Fd();
        return ret;
    }

    void SetConnectCallBack(const ConnectCallBack &cb)
    {
        _conn_call = cb;
    }

    void Listen()
    {
        _accept_channel.EnableRead();   
    }
    
};

class LoopThread
{
private:
    EventLoop *_loop;
    std::thread _thread;
    std::mutex _mutex;
    std::condition_variable _con_wait;

private:
    
    void ThreadEntry()
    {
        EventLoop loop;
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _loop = &loop;
            _con_wait.notify_all();
        }
        _loop->Loop();
    }

public:
    LoopThread()
        :_loop(nullptr),
        _thread(std::bind(&LoopThread::ThreadEntry, this))
    {}

    EventLoop *GetLoop()
    {
        EventLoop *loop;
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _con_wait.wait(_lock, [&](){return _loop != nullptr; });
            loop = _loop;
        }
        
        return loop;
    }
};

class ThreadPool
{
private:
    std::vector<LoopThread *> _threads;
    std::vector<EventLoop *> _loops;
    int _thread_count;
    int _next_index;

public:
    ThreadPool()
        :_thread_count(0),
        _next_index(0)
    {}

    void SetThreadCount(int count)
    {
        _thread_count = count;
    }

    void CreateThread()
    {
        _threads.resize(_thread_count);
        _loops.resize(_thread_count);
        for (int i = 0; i < _thread_count; i++)
        {
            _threads[i] = new LoopThread();
            _loops[i] = _threads[i]->GetLoop();
        }
    }

    EventLoop *GetNextLoop()
    {
        EventLoop *loop = _loops[_next_index];
        _next_index = (_next_index + 1) % _thread_count;
        return loop;
    }

};

class TcpServer
{
public:
    using MessageCallBack = std::function<void(const PtrConn &, Buffer *)>;
    using ConnectCallBack = std::function<void(const PtrConn &)>;
    using CloseCallBack = std::function<void(const PtrConn &)>;
    using AnyCallBack = std::function<void(const PtrConn &)>;

    MessageCallBack _message_callback;
    ConnectCallBack _connect_callback;
    CloseCallBack _closecallback;
    AnyCallBack _anycallback;
    CloseCallBack _serverclose_call;

private:
    ThreadPool _thread_pool;
    EventLoop _baseloop;    
    Acceptor _accept;
    int _conn_id_increment;
    bool _inactivedestroy;
    int _timeout;
    std::unordered_map<int, PtrConn> _conns;

private:
    void RemoveConnectionInbaseLoop(const PtrConn &ptr)
    {
        int id = ptr->GetId();
        auto it = _conns.find(id);
        if (it == _conns.end())
        {
            ERR_LOG("cannot find Connection, tecpserver fatal, exit");
            exit(0);
        }
        //
        _conns.erase(it);
    }

    void TimerAfterInbaseLoop(const TaskFunc &cb)
    {
        _baseloop.AddTimer(_conn_id_increment++, _timeout, cb);
    }

    void RemoveConnection(const PtrConn &ptr)
    {
        _baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInbaseLoop, this, ptr));
    }

    void NewConnection(int sockfd)
    {
        int index = _conn_id_increment++;
        PtrConn ptr = std::make_shared<Connection>(_thread_pool.GetNextLoop(), index, sockfd);
        _conns.insert({index, ptr});
        ptr->SetMessageCallBack(_message_callback);
        ptr->SetConnectCallBack(_connect_callback);
        ptr->SetCloseCallBack(_closecallback);
        ptr->SetAnyCallBack(_anycallback);
        ptr->SetServerCloseCallBack(std::bind(&TcpServer::RemoveConnection, this, std::placeholders::_1));
        if (_inactivedestroy) ptr->EnableInActiveDestroy(index, _timeout, std::bind(&Connection::ReleaseConnection, ptr.get()));
        ptr->Established();    //连接自己设置自己的各项事件就绪方法， tcpserver来设置建立连接前后的方法， 控制连接启动读监控
    }

public:
    TcpServer(uint16_t port)
        : _accept(&_baseloop, port),
        _conn_id_increment(0),
        _timeout(0),
        _inactivedestroy(false)
    {
        _accept.SetConnectCallBack(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));
        _accept.Listen();
        DBG_LOG("finish");
    }

    void SetMessageCallBack(const MessageCallBack &cb)
    {
        _message_callback = cb;
    }

    
    void SetConnectCallBack(const ConnectCallBack &cb)
    {
        _connect_callback = cb;
    }

    
    void SetCloseCallBack(const CloseCallBack &cb)
    {
        _closecallback = cb;
    }

    
    void SetAnyCallBack(const AnyCallBack &cb)
    {
        _anycallback = cb;
    }

    void EnableInactiveDestroy(int delay)
    {
        _timeout = delay;
        _inactivedestroy = true;   
    }
    void CreateThread(int count)
    {
        _thread_pool.SetThreadCount(count);
        _thread_pool.CreateThread();
    }

    void TimeAfter(const TaskFunc &cb)
    {
        _baseloop.RunInLoop(std::bind(&TcpServer::TimerAfterInbaseLoop, this, cb));
    }

    void Listen()
    {
        _baseloop.Loop();
    }
};

